package kafka

import (
	"github.com/Shopify/sarama"
)

//生产者对象池
type (
	ProduceService struct {
		Config *sarama.Config
		Addr   []string
	}
)

//生产者服务
func NewProduceService(addr []string) *ProduceService {

	this := &ProduceService{
		Addr: addr,
	}

	config := sarama.NewConfig()
	// 等待服务器所有副本都保存成功后的响应
	config.Producer.RequiredAcks = sarama.WaitForAll
	// 随机的分区类型：返回一个分区器，该分区器每次选择一个随机分区
	config.Producer.Partitioner = sarama.NewRandomPartitioner
	// 是否等待成功和失败后的响应
	config.Producer.Return.Successes = true

	config.Consumer.Return.Errors = true

	this.Config = config

	return this
}

//生成一个新的生产者
func (this *ProduceService) NewProducer() (producer sarama.SyncProducer, err error) {
	var client sarama.Client

	if client, err = sarama.NewClient(this.Addr, this.Config); err != nil {
		return
	}

	// 使用给定代理地址和配置创建一个同步生产者
	if producer, err = sarama.NewSyncProducerFromClient(client); err != nil {
		return
	}

	return
}
